/***************************************************************************

    MapCG: MapReduce Framework for CPU & GPU

    Copyright (C) 2010, Chuntao HONG (Chuntao.Hong@gmail.com).

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.

**************************************************************************/

#ifndef MAPREDUCE_H
#define MAPREDUCE_H

#include <stdio.h>

#if defined(HMMHASHTABLECPU_H)
#define HMM_FUNC HMMSchedulerCPU:: 
namespace HMM_CPU{
#elif defined(HMMHASHTABLEGPU_H)
#define HMM_FUNC __device__
namespace HMM_GPU{
#endif

void HMM_FUNC slice(const void * input_array, const unsigned int data_size, 
				const unsigned int unit_size, const unsigned int index, 
				const void * &ret_ptr, unsigned int & ret_size){
	default_slice(input_array, data_size, unit_size, index, ret_ptr, ret_size);
}

void HMM_FUNC map(const void * ptr, const unsigned int size, const global_data_t & global_data)
{
	relTuple * tuples = global_data.tuples;
	unsigned int * offsets = (unsigned int*)ptr;
	int numRecords = size / sizeof(int);
	for(int i = 0; i < numRecords; ++i)
	{
		relTuple & t = *(tuples + offsets[i]);
		emit_intermediate(&(t.record.key), sizeof(int), &t, sizeof(relTuple));
	}
}

void HMM_FUNC reduce(HMMKVs_t & kv, const global_data_t & global_data)
{
	// get the first value
	ValueListNode * currVal = kv.vit.ptr;
	ValueListNode * compVal;
	int numOutputs = 0;

	//// COUNT ////
	// if the current value is not null
	while (currVal != NULL)
	{
		// get the next value to compare against
		compVal = currVal->next;

		// if the comparison value is not null
		while (compVal != NULL)
		{
			// get the two tuples
			relTuple & t1 = *((relTuple *)(currVal->data.val));
			relTuple & t2 = *((relTuple *)(compVal->data.val));

			// if the table identifier for both values is different
			if (t1.relationId != t2.relationId)
			{
				++numOutputs;
			}

			// increment the comparison value
			compVal = compVal->next;
		}

		// increment the current value
		currVal = currVal->next;
	}
	
	// store this again first, as the false emit changes what kv.vit.ptr points to
	currVal = kv.vit.ptr;
	// false emit to allocate the space required for output data. It currently copies the input data instead
	emit(kv, kv.vit.ptr, numOutputs * sizeof(outputTupleNode) + sizeof(unsigned int) * 4); // the * 4 is a bad workaround, b/c mapcg is currently cutting the last 3 ints off my output in each emit :(

	// grab the start of the allocated array and store it locally
	unsigned int * const no = (unsigned int*)kv.kit.ptr->data.vlist.head->data.val;
	*no = numOutputs;

	//// MALLOC ////
	if (numOutputs != 0)
	{
		outputTupleNode * const outTuples = (outputTupleNode *)(no + sizeof(unsigned int));
		// reset number of outputs as it is used in the following logic
		numOutputs = 0;

		//// STORE ////
		// if the current value is not null
		while (currVal != NULL)
		{
			// get the next value to compare against
			compVal = currVal->next;

			// if the comparison value is not null
			while (compVal != NULL)
			{
				// get the two tuples
				relTuple & t1 = *((relTuple *)(currVal->data.val));
				relTuple & t2 = *((relTuple *)(compVal->data.val));

				// if the table identifier for both values is different
				if (t1.relationId != t2.relationId)
				{
					// perform a join and store the output in the out array

					// outTuples[numOutputs].aId = t1.relationId;
					// outTuples[numOutputs].bId = t2.relationId;
					// outTuples[numOutputs].aKey = t1.record.key;
					// outTuples[numOutputs].bKey = t2.record.key;
					outTuples[numOutputs].aData = t1.record.payload;
					outTuples[numOutputs].bData = t2.record.payload;
					++numOutputs;
				}

				// increment the comparison value
				compVal = compVal->next;
			}

			// increment the current value
			currVal = currVal->next;
		}

		// no need to emit results, as that has already been sneakily done
	}
}

void HMM_FUNC combine(void * old_val, const void * new_val){

}

};
#endif
